package com.guandata.cdc;

import com.guandata.cdc.model.db_type.gauss.GaussSinkConfig;
import com.guandata.cdc.model.db_type.postgres.PostgresSinkConfig;
import com.guandata.cdc.model.db_type.mysql.MysqlSourceConfig;
import com.guandata.cdc.model.db_type.postgres.PostgresSourceConfig;
import com.guandata.cdc.model.DataTableType;
import com.guandata.cdc.sink.impl.GaussSQLSinkFactory;
import com.guandata.cdc.sink.impl.PostgresSQLSinkFactory;
import com.guandata.cdc.source.GuanDataSourceFactory;
import com.guandata.cdc.source.impl.MysqlSQLSourceFactory;
import com.guandata.cdc.source.impl.PostgresSQLSourceFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.ExternalizedCheckpointRetention;
import org.apache.flink.core.execution.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.concurrent.TimeUnit;

/**
 * Flink CDC Biz Data Sync
 *
 * <pre>
 * <a href="https://blog.csdn.net/qq_41613913/article/details/145571445">Reference source code</a>
 * <a href="https://nightlies.apache.org/flink/flink-cdc-docs-master/docs/connectors/flink-sources/postgres-cdc/">flink sql data type</a>
 * </pre>
 */
@Slf4j
public class FlinkCDCBizDataSync {

    public static void main(String[] args) {
        // execution environment
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        var checkpointInterval = parameterTool.getLong(
            "checkpoint.interval.millis",
            TimeUnit.SECONDS.toMillis(30)
        );

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // set checkpoint
        env.enableCheckpointing(checkpointInterval);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointingConsistencyMode(CheckpointingMode.EXACTLY_ONCE);
        checkpointConfig.setExternalizedCheckpointRetention(ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION);
        checkpointConfig.setCheckpointTimeout(TimeUnit.MINUTES.toMillis(10));
        checkpointConfig.setMinPauseBetweenCheckpoints(TimeUnit.MILLISECONDS.toMillis(500));
        // max concurrent checkpoint = 1
        checkpointConfig.setMaxConcurrentCheckpoints(1);

        // table environment
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        var sourceTableType = DataTableType.from(parameterTool.getRequired("source.table.type"));
        log.info("source flink table type is: {}", sourceTableType);

        Table sourceTable;
        GuanDataSourceFactory<?> sourceFactory;
        switch(sourceTableType) {
            case POSTGRES:
                var postgresSourceConfig = new PostgresSourceConfig(parameterTool);
                var postgresSQLSourceFactory = new PostgresSQLSourceFactory(postgresSourceConfig);
                sourceTable = postgresSQLSourceFactory.registerSourceTable(tableEnv);
                sourceFactory = postgresSQLSourceFactory;
                break;
            case MYSQL:
                var mysqlSourceConfig = new MysqlSourceConfig(parameterTool);
                var mysqlSQLSourceFactory = new MysqlSQLSourceFactory(mysqlSourceConfig);
                sourceTable = mysqlSQLSourceFactory.registerSourceTable(tableEnv);
                sourceFactory = mysqlSQLSourceFactory;
                break;
            case GAUSS:
            default:
                throw new RuntimeException("Unsupported source table type: " + sourceTableType);
        }

        var sinkTableType = DataTableType.from(parameterTool.getRequired("sink.table.type"));
        log.info("sink flink table type is {}", sinkTableType);

        String sinkFlinkTableName;
        var sourceTableName = sourceFactory.getConfig().rawSourceTable;
        switch(sinkTableType) {
            case POSTGRES:
                var postgresSinkConfig = new PostgresSinkConfig(parameterTool);
                var postgresSQLSinkFactory = new PostgresSQLSinkFactory(postgresSinkConfig, sourceFactory.getSchema());
                sinkFlinkTableName = postgresSQLSinkFactory.registerSinkTable(tableEnv, sourceTableName);
                break;
            case GAUSS:
                var gaussSinkConfig = new GaussSinkConfig(parameterTool);
                var gaussSQLSinkFactory = new GaussSQLSinkFactory(gaussSinkConfig, sourceFactory.getSchema());
                sinkFlinkTableName = gaussSQLSinkFactory.registerSinkTable(tableEnv, sourceTableName);
                break;
            case MYSQL:
            default:
                throw new RuntimeException("Unsupported sink table type: " + sinkTableType);
        }

        sourceTable.executeInsert(sinkFlinkTableName);
        log.info("begin to sync data");
    }
}